package com.gjy.scheduling.monitor.processor;

import org.apache.commons.collections4.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.annotation.SchedulingConfigurer;
import org.springframework.scheduling.config.ScheduledTaskRegistrar;
import org.springframework.scheduling.config.TriggerTask;
import org.springframework.scheduling.support.CronTrigger;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;

/**
 * @author gjy
 * @version 1.0
 * @since 2025-08-30 21:12:17
 */
public abstract class AbstractDsHandler<T extends DsTaskInfo> implements SchedulingConfigurer {
    private static final Logger log = LoggerFactory.getLogger(AbstractDsHandler.class);

    private DsContainer<T> container;
    private final String className = getClass().getSimpleName();

    /**
     * 获取用于执行任务的线程池
     */
    protected abstract ExecutorService getWorkerExecutor();

    /**
     * 获取所有的任务信息
     */
    protected abstract List<T> listTaskInfo();

    /**
     * 做具体的任务逻辑
     */
    protected abstract void doProcess(T taskInfo);

    @Override
    public void configureTasks(ScheduledTaskRegistrar registrar) {
        container = new DsContainer<>(registrar);
        registrar.addFixedDelayTask(this::schedule, 1000);
    }

    /**
     * 调度任务，加载所有任务并注册
     * 首先加载所有任务信息，然后基于 cron 表达式生成 TriggerTask 对象，调用 checkTask() 方法确认是否需要注册/取消任务。
     * 当达到执行时间时，调用 execute() 方法，执行任务逻辑
     */
    private void schedule() {
        CollectionUtils.emptyIfNull(listTaskInfo())
                .forEach(t -> container.checkTask(t, new TriggerTask(() -> {
                    this.execute(t);
                }, context -> new CronTrigger(t.getCron()).nextExecutionTime(context))));
    }

    private void execute(T taskInfo) {
        String taskId = taskInfo.getId();
        try {
            Semaphore semaphore = container.getSemaphore(taskId);
            if (null == semaphore) {
                log.error("{} semaphore is null, taskId: {}", className, taskInfo);
                return;
            }
            if (semaphore.tryAcquire(3, TimeUnit.SECONDS)) {
                try {
                    getWorkerExecutor().execute(() -> doProcess(taskInfo));
                } finally {
                    semaphore.release();
                }
            } else {
                log.warn("{} too many executor, taskId: {}", className, taskId);
            }
        } catch (InterruptedException e) {
            log.warn("{} interruptedException error, taskId: {}", className, taskId);
            Thread.currentThread().interrupt();
        } catch (Exception e) {
            log.error("{} execute error, taskId: {}", className, taskId, e);
        }
    }
}
